package com.ry.flink.jobx;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.types.logical.DateType;
import org.apache.flink.types.Row;

import java.time.Instant;

/**
 *  7. 关于 toDataStream转换为DataStream显示的指定DataType
 * */
public class Job7 {

    public static class User {
        public String name;
        public Integer score;
        public Instant event_time;

    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tablEnv = StreamTableEnvironment.create(env);

        tablEnv.executeSql("CREATE TABLE userTable"
                + "("
                + " name STRING,"
                + " score INT,"
                + " event_time TIMESTAMP_LTZ(3),"
                + " WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND"
                + ")"
                + " WITH ('connector'='datagen')");
        Table userTable = tablEnv.from("userTable");
        /**
         * (
         *   `name` STRING,
         *   `score` INT,
         *   `event_time` TIMESTAMP_LTZ(3) *ROWTIME*,
         *   WATERMARK FOR `event_time`: TIMESTAMP_LTZ(3) AS `event_time` - INTERVAL '10' SECOND
         * )
         * */
        userTable.printSchema();

        DataStream<Row> dataStream1 = tablEnv.toDataStream(userTable);
        //基于反射的方式提取数据类型
        DataStream<User> dataStream2 = tablEnv.toDataStream(userTable,User.class);
//        dataStream.print();

        /**
         * 显示指定的数据类型
         * */
        tablEnv.toDataStream(userTable,DataTypes.STRUCTURED(
                User.class,
                DataTypes.FIELD("name", DataTypes.STRING()),
                DataTypes.FIELD("score", DataTypes.INT()),
                DataTypes.FIELD("event_time", DataTypes.TIMESTAMP_LTZ(3))));

        env.execute("job7");
    }
}
